package ThreadPool_26;


import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * 自定义实现阻塞队列线程池
 */
public class MyBlockThreadPool {

    // 线程池大小
    private final int poolSize;

    // 最大任务数
    private final int maxTaskCount;

    // 线程池线程列表
    private final List<PoolThread> poolThreadList;

    // 线程池里面的阻塞对了，存放线程对象
    private final MyBlockingQueue<Runnable> poolBlockingQueue;

    // 是否停止线程池工作
    private boolean isStopped = false;

    public MyBlockThreadPool(int poolSize, int maxTaskCount) {
        this.poolSize = poolSize;
        this.maxTaskCount = maxTaskCount;
        this.poolBlockingQueue = new MyBlockingQueue<Runnable>(maxTaskCount, "线程池任务队列");
        this.poolThreadList = new ArrayList<>();

        // 根据线程大小，创建线程对象
        for(int i = 1; i <= poolSize; i++) {
            poolThreadList.add(new PoolThread(poolBlockingQueue, i));
        }

        // 遍历启动线程
        poolThreadList.forEach(Thread::start);

    }

    /**
     * 执行
     * @param task
     */
    public synchronized void execute(Runnable task) throws InterruptedException {

        if(isStopped) {
            throw new IllegalStateException("线程池停止工作");
        }

        poolBlockingQueue.enqueue(task);
    }

    // 停止线程池工作
    public synchronized void stop() {
        this.isStopped = true;

        // 停止线程列表里面的所有线程
        poolThreadList.forEach(PoolThread::doStop);

        poolBlockingQueue.doStop();
    }

    /**
     * 线程池线程对象，从任务列表中获取任务，执行
     */
    public class PoolThread extends Thread {

        // 任务队列
        private final MyBlockingQueue<Runnable> taskQueue;

        // 控制线程是否停止
        private boolean isStopped = false;

        // 初始化时，将任务列表传入
        public PoolThread(MyBlockingQueue<Runnable> taskQueue, int threadNo) {
            this.setName("线程-" + threadNo);
            this.taskQueue = taskQueue;
        }

        // 定义多线程执行方法
        public void run() {
            while (!isStopped) {

                // 从任务队列获取一个任务执行
                Runnable dequeue = null;
                try {
                    dequeue = taskQueue.dequeue();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if(Objects.nonNull(dequeue)) {
                    dequeue.run();
                }
            }
        }

        // 停止线程执行
        public synchronized void doStop() {
            isStopped = true;

//            this.interrupt();
        }

        public synchronized boolean isStopped() {
            return isStopped;
        }
    }

    public static void main(String[] args) throws InterruptedException {


        MyBlockingQueue<String> myBlockingQueue = new MyBlockingQueue<String>(50, "数据队列");


        MyBlockThreadPool myBlockThreadPool = new MyBlockThreadPool(50, 50);

        long startTime = System.currentTimeMillis();
        System.out.println("任务开始：" + startTime);

        // 模拟10w条处理情况
        List<String> dataList = new ArrayList<>();
        for (int i = 1; i <= 100; i++) {
            dataList.add("数据-" + i);
        }
        for(String data : dataList) {
            // 获取数据进行处理
            myBlockThreadPool.execute(() -> {
                // 模拟做数据处理，包括数据清洗之类的，花费0.1秒
                try {
                    Thread.sleep(50);
                    myBlockingQueue.enqueue(data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

            // 模拟获取处理完后的数据，进行入口操作
            myBlockThreadPool.execute(() -> {
                // 模拟做数据入库操作，花费0.5秒
                try {
                    myBlockingQueue.dequeue();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        while (myBlockingQueue.getEnqueueTimes().get() != dataList.size()) {
//            System.out.println("aaa");
        }
        myBlockThreadPool.stop();
        System.out.println("任务结束：" + (System.currentTimeMillis() - startTime) / 1000);
    }
}
